package cn.ac.iscas.service.transformer;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.base.utils.TypeConvertUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author LJian
 * @version 1.0
 * @description:
 * {
 * 		"id": "增加常量列-2",
 * 		"alias": "增加常量列-2",
 * 		"entryClassName": "add_constant_columns",
 *         "type": "transformer",
 * 		"kind": "add_constant_columns",
 * 		"xPos": -373,
 * 		"yPos": 95,
 * 		"type": "transformer",
 * 		"front": ["MySQL-1"],
 * 		"next": ["过滤条件-3"],
 * 		"parameters": {
 *             "columns": [
 *                 {
 *                     "name": "c100",
 *                     "type": "String",
 *                     "defaultVal": "aa"
 *                 }
 *             ]
 *      }
 * }
 * @date 2022/7/14 14:21
 */
public class AddConstantColumnsTransformer {
    private final static String ID = "add_constant_column";
    private final static String COLUMNS_KEY = "columns";
    private final static String COLUMN_NAME_KEY = "name";
    private final static String COLUMN_TYPE_KEY = "type";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public AddConstantColumnsTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public AddConstantColumnsTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public AddConstantColumnsTransformer() {

    }

    //String sql = "CREATE VIEW middle_platform.db1.view2 AS (Select * , 'a' AS C100 from middle_platform.db1.iganalysis_vazu WHERE C1='1002' )";
    public String getSql(Map<String, Object> parameters) {
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        List<FlinkSourceColumn> columns = ParameterUtils.getRequiredListObj(parameters, COLUMNS_KEY, FlinkSourceColumn.class);

        StringBuffer constantColumn = new StringBuffer();
        for (FlinkSourceColumn column : columns) {
            if (column.getType().equalsIgnoreCase("string")) {
                constantColumn.append(String.format(" '%s' AS %s, ", column.getDefaultVal().toString(), column.getName()));
            } else {
                constantColumn.append(String.format(" %s AS %s, ", column.getDefaultVal().toString(), column.getName()));
            }
        }
        String constantColumnSchema = constantColumn.toString();
        constantColumnSchema = constantColumnSchema.substring(0, constantColumnSchema.lastIndexOf(","));
        String selectSql = String.format("( SELECT %s , %s FROM %s )", TypeConvertUtil.columnsNameToSelectString(this.frontColumns), constantColumnSchema, frontTableName);
        createViewSql += selectSql;

        //在之前列的基础上增加当前新增的常量列
        this.currentColumns.addAll(columns);
        updateSource();

        return createViewSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);

        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
